package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.util.KafkaUtil;
import com.atguigu.gmall.realtime.util.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author caodan
 * @version 1.0
 * @date 2022-09-28 14:29
 * 订单相关提前预处理
 */
public class DwdTradeOrderPreProcess {
    public static void main(String[] args) {

        // TODO 创建实时环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        /*// TODO 设置检查点相关设置
        // 开启检查点设置
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 设置超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        // 设置最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        // 设置 取消job 是否保留检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        // 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs:hadoop102:8020/xxxxx");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        //TODO 从kafka中获取相关数据
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "  `database` STRING,\n" +
                "  `table` STRING,\n" +
                "  `type` STRING,\n" +
                "  `data` map<string, string>,\n" +
                "  `old` map<string, string>,\n" +
                "  `ts` STRING,\n" +
                "  `proc_time` as PROCTIME()\n" +
                ") " + KafkaUtil.getKafkaDDLPorps("topic_db", "dwd_trand_order_pre_process"));

        // TODO 从db_topic 表中获取 order_detail ，order_info，order_detail_activity，order_detail_coupon
        Table orderDetail = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['order_id'] order_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['sku_name'] sku_name,\n" +
                "data['create_time'] create_time,\n" +
                "data['source_id'] source_id,\n" +
                "data['source_type'] source_type,\n" +
                "data['sku_num'] sku_num,\n" +
                "cast(cast(data['sku_num'] as decimal(16,2)) * " +
                "cast(data['order_price'] as decimal(16,2)) as String) split_original_amount,\n" +
                "data['split_total_amount'] split_total_amount,\n" +
                "data['split_activity_amount'] split_activity_amount,\n" +
                "data['split_coupon_amount'] split_coupon_amount,\n" +
                "ts od_ts,\n" +
                "proc_time \n" +
                "from `topic_db` where `table` = 'order_detail' " +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_detail", orderDetail);

        //获取order_info 数据
        Table orderInfo = tableEnv.sqlQuery("select\n" +
                "    data['id'] id,\n" +
                "    data['user_id'] user_id,\n" +
                "    data['operate_time'] operate_time,\n" +
                "    data['order_status'] order_status,\n" +
                "    data['province_id'] province_id,\n" +
                "    `type`,\n" +
                "    `old`,\n" +
                "    `ts` oi_ts\n" +
                "from\n" +
                "    topic_db\n" +
                "where \n" +
                "    `table` = 'order_info' and (`type` = 'insert' or `type` = 'update')");

        tableEnv.createTemporaryView("order_info", orderInfo);
        // 获取order_detail_activity 数据
        Table orderDetailActivity = tableEnv.sqlQuery("select \n" +
                "data['order_detail_id'] order_detail_id,\n" +
                "data['activity_id'] activity_id,\n" +
                "data['activity_rule_id'] activity_rule_id\n" +
                "from `topic_db`\n" +
                "where `table` = 'order_detail_activity'\n" +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_detail_activity", orderDetailActivity);

        //获取order_detail_coupon数据
        Table orderDetailCoupon = tableEnv.sqlQuery("select\n" +
                "data['order_detail_id'] order_detail_id,\n" +
                "data['coupon_id'] coupon_id\n" +
                "from `topic_db`\n" +
                "where `table` = 'order_detail_coupon'\n" +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_detail_coupon", orderDetailCoupon);

        // TODO 获取mysql字典表数据
        tableEnv.executeSql(MysqlUtil.getMysqlDDLStr());

        // TODO 将五张表数据 进行关联
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "od.id,\n" +
                "od.order_id,\n" +
                "od.sku_id,\n" +
                "od.sku_name,\n" +
                "date_format(od.create_time, 'yyyy-MM-dd') date_id,\n" +
                "od.create_time,\n" +
                "od.source_id,\n" +
                "od.source_type,\n" +
                "od.sku_num,\n" +
                "od.split_original_amount,\n" +
                "od.split_activity_amount,\n" +
                "od.split_coupon_amount,\n" +
                "od.split_total_amount,\n" +
                "od.od_ts,\n" +

                "oi.user_id,\n" +
                "oi.province_id,\n" +
                "date_format(oi.operate_time, 'yyyy-MM-dd') operate_date_id,\n" +
                "oi.operate_time,\n" +
                "oi.order_status,\n" +
                "oi.`type`,\n" +
                "oi.`old`,\n" +
                "oi.oi_ts,\n" +

                "act.activity_id,\n" +
                "act.activity_rule_id,\n" +

                "cou.coupon_id,\n" +

                "dic.dic_name source_type_name,\n" +

                "current_row_timestamp() row_op_ts\n" +
                "from order_detail od \n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id\n" +
                "left join order_detail_activity act\n" +
                "on od.id = act.order_detail_id\n" +
                "left join order_detail_coupon cou\n" +
                "on od.id = cou.order_detail_id\n" +
                "join `base_dic` for system_time as of od.proc_time as dic\n" +
                "on od.source_type = dic.dic_code");
        tableEnv.createTemporaryView("result_table", resultTable);
        // 创建kafka对应主题表
        tableEnv.executeSql("" +
                "create table dwd_trade_order_pre_process(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "sku_id string,\n" +
                "sku_name string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "source_id string,\n" +
                "source_type string,\n" +
                "sku_num string,\n" +
                "split_original_amount string,\n" +
                "split_activity_amount string,\n" +
                "split_coupon_amount string,\n" +
                "split_total_amount string,\n" +
                "od_ts string,\n" +

                "user_id string,\n" +
                "province_id string,\n" +
                "operate_date_id string,\n" +
                "operate_time string,\n" +
                "order_status string,\n" +
                "`type` string,\n" +
                "`old` map<string,string>,\n" +
                "oi_ts string,\n" +

                "activity_id string,\n" +
                "activity_rule_id string,\n" +

                "coupon_id string,\n" +

                "source_type_name string,\n" +

                "row_op_ts timestamp_ltz(3),\n" +
                "primary key(id) not enforced\n" +
                ")" + KafkaUtil.getUpsetKafkaProps("dwd_trade_order_pre_process"));

        // 将关联结果写入 Upsert-Kafka 表
        tableEnv.executeSql("" +
                "insert into dwd_trade_order_pre_process \n" +
                "select * from result_table");

    }
}
